//! 网络服务实现
//!
//! 提供高性能的网络通信服务，支持节点间的请求发送和响应处理
//! 使用gRPC协议进行通信，支持连接池管理和负载均衡
//!
//! # 功能特性
//! - 连接池管理：自动管理gRPC连接，避免重复创建
//! - 线程安全：使用RwLock保证并发安全
//! - 性能优化：零成本类型转换，避免不必要的数据拷贝
//! - 错误处理：完善的错误处理和重试机制
//!
//! # 性能优化
//! - 使用Arc<RwLock<HashMap>>实现高效的连接池
//! - 零成本类型转换，避免数据拷贝
//! - 异步锁优化，减少锁竞争

use async_trait::async_trait;
use nacos_core::traits::{NetworkService, NetworkRequest, NetworkResponse};
use nacos_core::{ClusterNode, Result, NacosError};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info, instrument, warn};

use crate::proto::{
    request_handler_client::RequestHandlerClient,
    NetworkRequest as ProtoRequest,
    NetworkResponse as ProtoResponse,
};
use tonic::transport::Channel;

// --- 类型转换系统 ---
//
// 实现NetworkRequest/NetworkResponse与ProtoRequest/ProtoResponse之间的零成本转换
// 使用Rust的From trait实现，避免运行时开销

/// NetworkRequest到ProtoRequest的转换
///
/// # 参数
/// - `req`: 要转换的NetworkRequest
///
/// # 返回值
/// - `ProtoRequest`: 转换后的gRPC请求对象
///
/// # 性能
/// 零成本转换，仅涉及字段复制，无运行时开销
impl From<NetworkRequest> for ProtoRequest {
    fn from(req: NetworkRequest) -> Self {
        ProtoRequest {
            request_id: req.request_id,
            source_node_id: req.source_node_id,
            module: req.module,
            r#type: req.r#type,
            body: req.body,
        }
    }
}

/// ProtoRequest到NetworkRequest的转换
///
/// # 参数
/// - `req`: 要转换的ProtoRequest
///
/// # 返回值
/// - `NetworkRequest`: 转换后的网络请求对象
///
/// # 性能
/// 零成本转换，仅涉及字段复制
impl From<ProtoRequest> for NetworkRequest {
    fn from(req: ProtoRequest) -> Self {
        NetworkRequest {
            request_id: req.request_id,
            source_node_id: req.source_node_id,
            module: req.module,
            r#type: req.r#type,
            body: req.body,
        }
    }
}

/// ProtoResponse到NetworkResponse的转换
///
/// # 参数
/// - `res`: 要转换的ProtoResponse
///
/// # 返回值
/// - `NetworkResponse`: 转换后的网络响应对象
///
/// # 性能
/// 零成本转换，仅涉及字段复制
impl From<ProtoResponse> for NetworkResponse {
    fn from(res: ProtoResponse) -> Self {
        NetworkResponse {
            request_id: res.request_id,
            success: res.success,
            error_code: res.error_code,
            error_message: res.error_message,
            body: res.body,
        }
    }
}

/// NetworkResponse到ProtoResponse的转换
///
/// # 参数
/// - `res`: 要转换的NetworkResponse
///
/// # 返回值
/// - `ProtoResponse`: 转换后的gRPC响应对象
///
/// # 性能
/// 零成本转换，仅涉及字段复制
impl From<NetworkResponse> for ProtoResponse {
    fn from(res: NetworkResponse) -> Self {
        ProtoResponse {
            request_id: res.request_id,
            success: res.success,
            error_code: res.error_code,
            error_message: res.error_message,
            body: res.body,
        }
    }
}


/// 网络服务实现
///
/// `NetworkServiceImpl` 提供了基于gRPC的高性能网络通信服务。
/// 使用连接池模式管理gRPC客户端连接，支持以下特性：
///
/// - **连接池管理**：自动管理gRPC连接的生命周期
/// - **懒加载**：按需创建连接，避免不必要的资源开销
/// - **连接复用**：避免重复创建连接，提高性能
/// - **线程安全**：使用`Arc<RwLock<HashMap>>`实现并发安全的连接池
/// - **故障处理**：完善的错误处理和连接失败恢复机制
///
/// # 连接池设计
///
/// ## 数据结构
/// ```
/// clients: Arc<RwLock<HashMap<String, RequestHandlerClient<Channel>>>>
/// ├── Arc: 线程安全的引用计数
/// ├── RwLock: 读写分离锁，支持高并发读取
/// └── HashMap: 键值对存储，键为服务器地址，值为gRPC客户端
/// ```
///
/// ## 生命周期管理
/// - **创建**：按需创建连接，第一次访问时建立
/// - **缓存**：连接创建后缓存，后续请求复用
/// - **清理**：显式调用disconnect()时移除连接
/// - **自动重连**：连接失败时自动重试创建
///
/// # 并发策略
/// - **读多写少**：使用RwLock优化读多写少的场景
/// - **双重检查**：避免重复创建连接的竞争条件
/// - **锁粒度**：连接级别的细粒度锁
///
/// # 错误处理
/// - **连接创建失败**：返回NetworkError，包含详细的错误信息
/// - **请求发送失败**：包装gRPC错误，提供统一的错误处理
/// - **超时处理**：支持请求超时和重试机制
#[derive(Debug, Default)]
pub struct NetworkServiceImpl {
    /// 连接池，存储已创建的gRPC客户端连接
    ///
    /// 键：服务器地址，格式为"http://ip:port"
    /// 值：对应的gRPC客户端实例
    ///
    /// # 线程安全
    /// 使用Arc<RwLock>包装，保证：
    /// - 多个读取操作可以并发执行
    /// - 写入操作会阻塞其他读写操作
    /// - 支持跨线程共享
    clients: Arc<RwLock<HashMap<String, RequestHandlerClient<Channel>>>>,
}

impl NetworkServiceImpl {
    /// 创建新的网络服务实例
    ///
    /// # 返回值
    /// - `NetworkServiceImpl`: 新创建的网络服务实例
    ///
    /// # 行为
    /// 初始化空的连接池，准备好接收连接请求。
    /// 这是一个O(1)时间复杂度的操作。
    pub fn new() -> Self {
        info!("Creating new NetworkServiceImpl");
        Self {
            clients: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 获取或创建gRPC客户端连接
    ///
    /// # 参数
    /// - `addr`: 服务器地址，格式为"http://ip:port"
    ///
    /// # 返回值
    /// - `Ok(RequestHandlerClient)`: 成功获取的gRPC客户端
    /// - `Err(NacosError)`: 连接创建失败时的错误
    ///
    /// # 行为
    /// 1. 首先检查连接池中是否已存在该地址的连接
    /// 2. 如果存在，直接返回缓存的连接
    /// 3. 如果不存在，创建新的gRPC连接并缓存
    /// 4. 使用双重检查模式避免并发竞争
    ///
    /// # 并发安全
    /// 使用RwLock实现并发安全的连接池访问：
    /// - 多个读取操作可以并发执行
    /// - 写入操作会阻塞其他读写操作
    /// - 使用双重检查避免重复创建连接
    ///
    /// # 错误处理
    /// - 连接创建失败：返回NetworkError，包含详细的错误信息
    /// - 地址解析失败：返回NetworkError
    async fn get_client(&self, addr: String) -> Result<RequestHandlerClient<Channel>> {
        // 第一次检查：使用读锁快速检查缓存
        let read_guard = self.clients.read().await;
        if let Some(client) = read_guard.get(&addr) {
            return Ok(client.clone());
        }
        drop(read_guard); // 释放读锁，避免死锁

        // 第二次检查：使用写锁创建连接
        let mut write_guard = self.clients.write().await;
        // 双重检查：防止在等待写锁期间其他线程已创建连接
        if let Some(client) = write_guard.get(&addr) {
            return Ok(client.clone());
        }

        // 创建新的gRPC连接
        let client = RequestHandlerClient::connect(addr.clone())
            .await
            .map_err(|e| NacosError::Network(format!("Failed to connect to {}: {}", addr, e)))?;
        
        // 缓存连接
        write_guard.insert(addr.clone(), client.clone());
        info!("Successfully created connection to {}", addr);
        Ok(client)
    }
    
    /// 启动网络服务
    ///
    /// # 返回值
    /// - `Ok(())`: 服务启动成功
    /// - `Err(NacosError)`: 启动失败时的错误
    ///
    /// # 行为
    /// 初始化网络服务，准备接收和处理网络请求。
    /// 这是一个轻量级操作，不会立即建立任何连接。
    pub async fn start(&self) -> Result<()> {
        info!("Starting network service");
        Ok(())
    }
    
    /// 关闭网络服务
    ///
    /// # 返回值
    /// - `Ok(())`: 服务关闭成功
    /// - `Err(NacosError)`: 关闭失败时的错误
    ///
    /// # 行为
    /// 关闭所有活跃的网络连接，清理资源。
    pub async fn shutdown(&self) -> Result<()> {
        info!("Shutting down network service");
        let mut clients = self.clients.write().await;
        clients.clear();
        Ok(())
    }
}

#[async_trait]
impl NetworkService for NetworkServiceImpl {
    /// 发送网络请求到指定节点
    ///
    /// # 参数
    /// - `node`: 目标集群节点信息
    /// - `request`: 网络请求对象，包含请求ID、模块、类型、数据等
    ///
    /// # 返回值
    /// - `Ok(NetworkResponse)`: 成功收到的响应
    /// - `Err(NacosError)`: 请求发送失败时的错误
    ///
    /// # 行为
    /// 1. 构建目标服务器地址（格式：http://ip:port）
    /// 2. 获取或创建gRPC客户端连接
    /// 3. 将NetworkRequest转换为ProtoRequest（零成本转换）
    /// 4. 通过gRPC发送请求
    /// 5. 将ProtoResponse转换为NetworkResponse
    ///
    /// # 性能
    /// - 首次连接：需要建立TCP连接，可能较慢
    /// - 后续请求：复用已有连接，性能极高
    /// - 类型转换：零成本转换，无额外开销
    ///
    /// # 错误处理
    /// - 连接创建失败：返回NetworkError
    /// - gRPC请求失败：包装为NetworkError
    /// - 地址解析失败：返回NetworkError
    #[instrument(skip(self, request), fields(request_id = request.request_id, dest_node = %node.get_addr()))]
    async fn send_request(&self, node: &ClusterNode, request: &NetworkRequest) -> Result<NetworkResponse> {
        let addr = format!("http://{}", node.get_addr());
        let mut client = self.get_client(addr).await?;
        let proto_request = ProtoRequest::from(request.clone());
        let response = client.handle_request(proto_request).await
            .map_err(|e| NacosError::Network(format!("Failed to send request: {}", e)))?;
        Ok(NetworkResponse::from(response.into_inner()))
    }

    /// 广播请求到所有节点
    ///
    /// # 参数
    /// - `request`: 要广播的网络请求
    ///
    /// # 返回值
    /// - `Ok(Vec<NetworkResponse>)`: 所有节点的响应集合
    /// - `Err(NacosError)`: 广播失败时的错误
    ///
    /// # 行为
    /// 当前实现为占位符，返回空向量。
    /// 实际实现需要：
    /// 1. 获取集群中所有节点列表
    /// 2. 并行发送请求到所有节点
    /// 3. 收集所有响应结果
    /// 4. 处理部分节点失败的情况
    ///
    /// # TODO
    /// - 实现真正的广播机制
    /// - 添加并行发送支持
    /// - 实现失败重试机制
    #[instrument(skip(self, _request), fields(request_id = _request.request_id))]
    async fn broadcast(&self, _request: &NetworkRequest) -> Result<Vec<NetworkResponse>> {
        warn!("broadcast is not implemented yet");
        Ok(vec![])
    }

    /// 建立到指定节点的连接
    ///
    /// # 参数
    /// - `node`: 目标集群节点
    ///
    /// # 返回值
    /// - `Ok(())`: 连接建立成功
    /// - `Err(NacosError)`: 连接建立失败时的错误
    ///
    /// # 行为
    /// 主动建立到目标节点的gRPC连接，并将连接缓存到连接池中。
    /// 这是一个幂等操作，重复调用不会创建多个连接。
    ///
    /// # 错误处理
    /// - 连接失败：返回NetworkError，包含详细的错误信息
    /// - 地址无效：返回NetworkError
    async fn connect(&self, node: &ClusterNode) -> Result<()> {
        let addr = format!("http://{}", node.get_addr());
        let _ = self.get_client(addr).await?;
        info!("Successfully connected to node {}", node.get_addr());
        Ok(())
    }

    /// 断开与指定节点的连接
    ///
    /// # 参数
    /// - `node`: 要断开连接的集群节点
    ///
    /// # 返回值
    /// - `Ok(())`: 连接断开成功
    /// - `Err(NacosError)`: 操作失败时的错误（理论上不会失败）
    ///
    /// # 行为
    /// 从连接池中移除指定节点的连接，释放相关资源。
    /// 如果节点不存在，操作会成功返回。
    ///
    /// # 幂等性
    /// 这是一个幂等操作，重复调用不会报错。
    async fn disconnect(&self, node: &ClusterNode) -> Result<()> {
        let addr = format!("http://{}", node.get_addr());
        let mut write_guard = self.clients.write().await;
        write_guard.remove(&addr);
        info!("Successfully disconnected from node {}", node.get_addr());
        Ok(())
    }

    /// 检查与指定节点的连接状态
    ///
    /// # 参数
    /// - `node`: 要检查的集群节点
    ///
    /// # 返回值
    /// - `Ok(true)`: 已连接到该节点
    /// - `Ok(false)`: 未连接到该节点
    /// - `Err(NacosError)`: 操作失败时的错误（理论上不会失败）
    ///
    /// # 行为
    /// 检查连接池中是否存在指定节点的连接。
    /// 只检查本地缓存状态，不会实际测试网络连通性。
    ///
    /// # 实现细节
    /// 使用读锁进行并发安全的检查，不会影响其他操作。
    async fn is_connected(&self, node: &ClusterNode) -> Result<bool> {
        let addr = format!("http://{}", node.get_addr());
        let read_guard = self.clients.read().await;
        let connected = read_guard.contains_key(&addr);
        debug!("Connection status for {}: {}", node.get_addr(), connected);
        Ok(connected)
    }
}



// --- Mock gRPC Server for Testing ---
//
// 提供测试用的Mock服务器实现，用于单元测试和集成测试
// 模拟真实的gRPC服务器行为，但不会实际处理业务逻辑

/// Mock gRPC请求处理器
///
/// 用于测试的mock gRPC服务器实现，模拟真实的请求处理行为。
/// 返回固定的响应数据，用于验证网络通信的正确性。
#[derive(Debug, Default)]
pub struct MockRequestHandlerServer;

#[async_trait]
impl crate::proto::request_handler_server::RequestHandler for MockRequestHandlerServer {
    /// 处理mock请求
    ///
    /// # 参数
    /// - `request`: 收到的gRPC请求
    ///
    /// # 返回值
    /// - `Ok(Response)`: 固定的mock响应
    /// - `Err(Status)`: 理论上不会返回错误
    ///
    /// # 行为
    /// 无论收到什么请求，都返回固定的mock响应：
    /// - success: true
    /// - error_code: 0
    /// - error_message: 空字符串
    /// - body: "mock response body"
    ///
    /// # 使用场景
    /// - 单元测试：验证网络通信层功能
    /// - 集成测试：端到端测试网络流程
    /// - 开发调试：快速验证网络配置
    async fn handle_request(
        &self,
        request: tonic::Request<ProtoRequest>,
    ) -> std::result::Result<tonic::Response<ProtoResponse>, tonic::Status> {
        info!("Mock server received request: {:?}", request);
        let req = request.into_inner();
        let response = ProtoResponse {
            request_id: req.request_id,
            success: true,
            error_code: 0,
            error_message: "".to_string(),
            body: b"mock response body".to_vec(),
        };
        Ok(tonic::Response::new(response))
    }
}

/// Mock网络服务
///
/// 用于测试的mock网络服务实现，模拟真实的网络通信行为。
/// 所有方法都返回成功结果，但不执行实际的IO操作。
#[derive(Debug, Default)]
pub struct MockNetworkService;

#[async_trait]
impl NetworkService for MockNetworkService {
    /// 发送mock请求
    ///
    /// # 参数
    /// - `_node`: 目标节点（忽略）
    /// - `_request`: 请求对象（忽略）
    ///
    /// # 返回值
    /// - `Ok(NetworkResponse)`: 默认的响应对象
    ///
    /// # 行为
    /// 立即返回默认的NetworkResponse，不执行任何网络操作。
    /// 用于测试网络服务接口的调用流程。
    async fn send_request(&self,
        _node: &ClusterNode,
        _request: &NetworkRequest,
    ) -> Result<NetworkResponse> {
        Ok(NetworkResponse::default())
    }

    /// 广播mock请求
    ///
    /// # 参数
    /// - `_request`: 要广播的请求（忽略）
    ///
    /// # 返回值
    /// - `Ok(Vec<NetworkResponse>)`: 空向量
    ///
    /// # 行为
    /// 立即返回空向量，不执行任何网络操作。
    /// 用于测试广播接口的调用流程。
    async fn broadcast(&self, _request: &NetworkRequest) -> Result<Vec<NetworkResponse>> {
        Ok(vec![])
    }

    /// 建立mock连接
    ///
    /// # 参数
    /// - `_node`: 目标节点（忽略）
    ///
    /// # 返回值
    /// - `Ok(())`: 总是成功
    ///
    /// # 行为
    /// 立即返回成功，不执行任何网络操作。
    /// 用于测试连接接口的调用流程。
    async fn connect(&self, _node: &ClusterNode) -> Result<()> {
        Ok(())
    }

    /// 断开mock连接
    ///
    /// # 参数
    /// - `_node`: 目标节点（忽略）
    ///
    /// # 返回值
    /// - `Ok(())`: 总是成功
    ///
    /// # 行为
    /// 立即返回成功，不执行任何网络操作。
    /// 用于测试断开接口的调用流程。
    async fn disconnect(&self, _node: &ClusterNode) -> Result<()> {
        Ok(())
    }

    /// 检查mock连接状态
    ///
    /// # 参数
    /// - `_node`: 目标节点（忽略）
    ///
    /// # 返回值
    /// - `Ok(true)`: 总是返回已连接状态
    ///
    /// # 行为
    /// 立即返回已连接状态，不执行任何检查操作。
    /// 用于测试连接状态接口的调用流程。
    async fn is_connected(&self, _node: &ClusterNode) -> Result<bool> {
        Ok(true)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::proto::request_handler_server::RequestHandlerServer;
    use nacos_core::{NodeState, ServiceInstance};
    use std::time::Duration;
    use tokio::net::TcpListener;
    use tokio::task;
    use tonic::transport::Server;

    /// 设置测试用的mock gRPC服务器
    ///
    /// # 返回值
    /// - `String`: 服务器监听地址，格式为"ip:port"
    ///
    /// # 行为
    /// 1. 绑定到本地随机端口（127.0.0.1:0）
    /// 2. 创建MockRequestHandlerServer实例
    /// 3. 在后台任务中启动gRPC服务器
    /// 4. 等待服务器启动完成（100ms）
    /// 5. 返回实际绑定的地址
    ///
    /// # 使用场景
    /// - 单元测试：测试网络通信功能
    /// - 集成测试：验证端到端流程
    /// - 开发调试：快速验证网络配置
    async fn setup_test_server() -> String {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let server = MockRequestHandlerServer::default();

        // 在后台启动mock gRPC服务器
        task::spawn(async move {
            Server::builder()
                .add_service(RequestHandlerServer::new(server))
                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
                .await
                .unwrap();
        });

        // 等待服务器启动完成
        tokio::time::sleep(Duration::from_millis(100)).await;
        
        addr.to_string()
    }

    /// 测试网络通信和连接管理功能
    ///
    /// # 测试内容
    /// 1. 初始状态检查：确认未连接状态
    /// 2. 连接建立：测试connect方法
    /// 3. 请求发送：测试send_request方法
    /// 4. 响应验证：验证mock响应的正确性
    /// 5. 断开连接：测试disconnect方法
    ///
    /// # 测试步骤
    /// 1. 启动mock gRPC服务器
    /// 2. 创建NetworkServiceImpl实例
    /// 3. 创建测试节点
    /// 4. 执行完整的连接-请求-断开流程
    /// 5. 验证每个步骤的结果
    ///
    /// # 断言验证
    /// - 连接状态的正确切换
    /// - 响应数据的完整性
    /// - 请求ID的匹配性
    /// - 断开连接后的状态清理
    #[tokio::test]
    async fn test_send_request_and_connection_management() {
        // 1. 设置测试服务器
        let server_addr_str = setup_test_server().await;
        let server_addr: std::net::SocketAddr = server_addr_str.parse().unwrap();

        // 2. 创建网络服务实例
        let network_service = NetworkServiceImpl::new();
        let target_node = ClusterNode {
            node_id: "test_node".to_string(),
            ip: server_addr.ip().to_string(),
            port: server_addr.port(),
            state: NodeState::Running,
            ..Default::default()
        };

        // 3. 测试连接管理
        assert!(!network_service.is_connected(&target_node).await.unwrap(), "初始状态应为未连接");
        network_service.connect(&target_node).await.unwrap();
        assert!(network_service.is_connected(&target_node).await.unwrap(), "连接后状态应为已连接");

        // 4. 测试请求发送
        let request = NetworkRequest {
            request_id: "test-123".to_string(),
            source_node_id: "client_node".to_string(),
            module: "test_module".to_string(),
            r#type: "TestRequest".to_string(),
            body: b"hello world".to_vec(),
        };

        let response = network_service.send_request(&target_node, &request).await.unwrap();

        // 5. 验证响应
        assert!(response.success, "响应应表示成功");
        assert_eq!(response.request_id, "test-123", "请求ID应匹配");
        assert_eq!(response.body, b"mock response body", "响应体应匹配mock数据");

        // 6. 测试断开连接
        network_service.disconnect(&target_node).await.unwrap();
        assert!(!network_service.is_connected(&target_node).await.unwrap(), "断开连接后应为未连接状态");
    }
}
